/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package mapreduce;

/**
 *
 * @author haoyun
 */

import cfar.ff.fun.F2;
import cfar.ff.fun.F1;
import cfar.ff.interfaces.Monoid;
import mapreduce.impMono;
import java.util.Iterator;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.lang.Thread;
import static fj.data.List.*;
import fj.data.List;

public class RReduce<A> {

    public A reduce(List<A> data, F2<A,A,A> func, A empt){

        List<Integer> re=list();
        for (int i=0;i<data.length();i++){
            re=re.snoc(1);
        }
        return this.reduce(data, func, empt,re);
    }

    public A reduce(List<A> data, F2<A,A,A> func, A empt, List<Double> rel, List<Double> err){

        List <Integer> num=Reliable.prob2num(rel,err);
        return this.reduce(data, func, empt,num);
    }

    public A reduce(List<A> data, F2<A,A,A> func, A empt, List<Integer> re){

        List<A> reducedlist=data;
        List<A> curr=list();
        int leng=reducedlist.length();
        int id=0;
        int relen=0;
        for(int i=0;i<re.length();i++){
            relen+=re.index(i);
        }
        runReduce[] runner=new runReduce[relen];
        while(leng>1){
            id=0;
            for (int i=0;i<leng;i++){
                if(i+1==leng){
                    for (int j=0;j<re.index(i);j++){
                        runner[id]=new runReduce();
                        runner[id].setresult(reducedlist.index(i));
                        id++;
                    }
                    re=re.snoc(re.index(i));
                }
                else{
                    curr=list(reducedlist.index(i),reducedlist.index(i+1));
                    for (int j=0;j<Math.max(re.index(i), re.index(i+1));j++){
                        runner[id]=new runReduce(curr,func,empt);
                        id++;
                    }
                    re=re.snoc(Math.max(re.index(i), re.index(i+1)));
                }
                i=i+1;
            }

            //check whether runner are all completed
            boolean complete=false;
            while(!complete){
                complete=true;
                for(int i=0;i<id;i++){
                    if(!runner[i].finished)
                        complete=false;
                }
                try{
                    Thread.sleep(1000);
                }catch (InterruptedException ie){
                    //do nothing
                }
            }
            
            re=re.drop(leng);
            
            reducedlist=list();
            int reid=0;
            int i=0;
            while (i<id){
                int num=re.index(reid);
                boolean goodresult=false;
                for(int j=0;j<num;j++){
                    if(!goodresult){
                        if(runner[i].getresult()!=null){
                            reducedlist=reducedlist.snoc((A)runner[i].getresult());
                            i++;
                            goodresult=true;
                            continue;
                        }
                        else{
                            i++;
                        }
                    }
                    else{
                        i++;
                    }
                }
                reid++;
            }
            leng=reducedlist.length();
        }
        return reducedlist.index(0);
     }


     //TEST!!!
    public static void main(String[] arg){
        F2 add10 = new F2<Integer,Integer,Integer>() {

        public Integer f(Integer a,Integer b) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException ex) {
                    Logger.getLogger(RMap.class.getName()).log(Level.SEVERE, null, ex);
                }
               return a+b;
            }
        };

        List<Integer> input=list(1,2,3,4,5,6,7,8,9,10,11);
        List<Integer> re=list(1,3,3,1,1,1,1,5,1,3,1,1);
        RReduce test=new RReduce();
        Integer empt=0;
        Object result=test.reduce(input, add10, empt);
        System.out.println(result);
    }

}

class runReduce<A> implements Runnable{

    List<A> data;
    F2<A,A,A> func;
    A empt;
    A result;
    private Thread runner;
    boolean finished=false;

    runReduce(){
        
    }

    runReduce(List<A> data,F2<A,A,A> func,A empt){
        this.data=data;
        this.func=func;
        this.empt=empt;
        if (runner==null){
            runner=new Thread(this);
            runner.start();
        }
    }

    public void run() {

        //Iterable<impMono<A>> Data=this.buildData(data, func, empt);
        //F1<Iterable<impMono<A>>,A> freduce=Data.iterator().next().concat();
        //result=freduce.f(Data);

        if(data.length()==1){
            result=data.index(0);
            finished=true;
        }
        else{
            result=func.f(data.index(0), data.index(1));
            finished=true;
        }

    }

    private Iterable<impMono<A>> buildData (List<A> data, F2<A,A,A> func, A empt){

        impMono<A> emptMono=new impMono(empt,func,empt);
        List<impMono<A>> Data=list(emptMono);
        for (Object o: data){
            impMono<A> Mono=new impMono(empt,func,o);
            Data=Data.snoc(Mono);

        }
        return Data;
    }

    public A getresult(){
        return result; 
    }

    public void setresult(A result){
        this.result=result;
        finished=true;
    }

}
